package io.github.rascaler.dcenter.tasks;

import com.alibaba.fastjson.JSON;
import com.binance.connector.futures.client.exceptions.BinanceClientException;
import com.binance.connector.futures.client.exceptions.BinanceConnectorException;
import com.binance.connector.futures.client.impl.UMFuturesClientImpl;
import com.binance.connector.futures.client.utils.ProxyAuth;
import io.github.rascaler.dcenter.constant.ConstantItem;
import io.github.rascaler.dcenter.db.DbManager;
import io.github.rascaler.dcenter.manager.BinanceManager;
import io.github.rascaler.dcenter.platforms.DbMapper;
import io.github.rascaler.dcenter.platforms.Frequency;
import io.github.rascaler.dcenter.platforms.Parser;
import io.github.rascaler.dcenter.platforms.binance.*;
import io.github.rascaler.dcenter.pojo.Candlestick;
import io.github.rascaler.dcenter.pojo.FixedArrayList;
import io.github.rascaler.dcenter.properties.BinanceProperties;
import io.github.rascaler.dcenter.properties.ProxyInfo;
import io.github.rascaler.dcenter.ws.handler.KlineSubscribeMessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.ref.Reference;
import java.math.BigDecimal;
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 交易查询任务
 * 主动去第三方支付平台查询交易记录
 */
@Slf4j
@Component
public class BinanceBarTask {

    @Autowired
    private DbManager dbManager;

    @Autowired
    private KlineSubscribeMessageHandler klineSubscribeMessageHandler;

    private final static ExecutorService pool = Executors.newFixedThreadPool(100);

    public final static UMFuturesClientImpl client = new UMFuturesClientImpl();

    @Autowired
    private BinanceProperties binanceProperties;

    @Resource(type = BinanceDbMapperImpl.class)
    private DbMapper dbMapper;

    @Resource(type = BinanceCandlestickParser.class)
    private Parser binanceCandlestickParser;

    @Resource(type = BinanceInstrumentParser.class)
    private Parser binanceInstrumentParser;


    private List<String> getNotLatestSymbols() {
        Map<String, Long> map = dbMapper.getTimeMap(ConstantItem.FREQUENCY_1MIN);
        List<String> symbols = new ArrayList<>();
        Long now = new Date().getTime();
        // 如果数据采集只要在8小时内都可以,超过8小时，数据可能就不连续了
        map.forEach((key, value) -> {
            if (null != value && now - value < binanceProperties.getExpireTime()) {
                symbols.add(key);
            }
        });
        return symbols;
    }





    public List<Candlestick> resample(List<Candlestick> originalData, long targetInterval) {
        List<Candlestick> resampledData = new ArrayList<>();

        // 计算目标间隔内的时间戳差值
        // 初始化目标时间戳
        long targetTimestamp = originalData.get(0).getBarDate();
        long endTime = originalData.get(originalData.size() - 1).getBarDate();

        while (targetTimestamp <= endTime) {
            // 在原始数据中找到在目标间隔内的所有蜡烛图数据
            List<Candlestick> candlesInInterval = new ArrayList<>();

            for (Candlestick candle : originalData) {
                if (candle.getBarDate() >= targetTimestamp && candle.getBarDate() < targetTimestamp + targetInterval) {
                    candlesInInterval.add(candle);
                }
            }

            // 如果存在蜡烛图数据，则计算目标间隔内的开盘价、最高价、最低价和收盘价，并添加到重新采样数据中
            if (!candlesInInterval.isEmpty()) {
                BigDecimal open = candlesInInterval.get(0).getOpen();
                BigDecimal high = BigDecimal.ZERO;
                BigDecimal low = BigDecimal.valueOf(Double.MAX_VALUE);
                BigDecimal close = candlesInInterval.get(candlesInInterval.size() - 1).getClose();
                BigDecimal volume = BigDecimal.ZERO;
                BigDecimal amount = BigDecimal.ZERO;

                for (Candlestick candle : candlesInInterval) {
                    high = high.max(candle.getHigh());
                    low = low.min(candle.getLow());
                    volume = volume.add(candle.getVolume());
                    amount = amount.add(candle.getAmount());
                }

                Candlestick resampledCandle = new Candlestick(targetTimestamp, open, high, low, close, volume, amount);
                resampledData.add(resampledCandle);
            }

            // 更新目标时间戳
            targetTimestamp += targetInterval;
        }

        return resampledData;
    }

    public List<Candlestick>  resample(String symbol, String frequency) {
        // OHLCV
        Long start = System.currentTimeMillis();
        String key = String.format(BinanceConstant.keyTemplate, symbol, ConstantItem.FREQUENCY_1MIN, ConstantItem.ADJUST_NO, ConstantItem.EXCHANGE_BINANCE);
        // 取出最后一条数据
        Map<String, Long> timeMap = dbMapper.getTimeMap(frequency);
        if (timeMap.size() == 0) {
            return new ArrayList<>();
        }
        Long lastTime = timeMap.get(symbol);
        FixedArrayList<Candlestick> klines = dbManager.getMdb().getBarMap().get(key);
        if (CollectionUtils.isEmpty(klines)) {
            return new ArrayList<>();
        }
        List<Candlestick> mergeKlines = klines.stream().filter(k -> k.getBarDate() >= lastTime).collect(Collectors.toList());
        List<Candlestick> result = resample(mergeKlines, Frequency.getFrequencyUnit(frequency));
//        log.error("resample，共{}条数据，耗时：{}ms", result.size(),System.currentTimeMillis() - start);
        return result;
    }

    public List<Candlestick>  batchResample() {
        Long start = System.currentTimeMillis();
        List<Candlestick> result = new ArrayList<>();
        Map<String, List<Candlestick>> map = new HashMap<>();
        List<Future<Map<String, List<Candlestick>>>> tasks = new ArrayList<>();
        binanceProperties.getFrequencies().forEach(f -> {
            if (f.equals(ConstantItem.FREQUENCY_1MIN)) {
                return;
            }
            tasks.add(pool.submit(() ->{
                Map<String, List<Candlestick>> single = new HashMap<>();
                dbManager.getMdb().getInstrumentMap().forEach((key, val) -> {
                    String smallKey = String.format(BinanceConstant.keyTemplate, val.getSymbol(), ConstantItem.FREQUENCY_1MIN, ConstantItem.ADJUST_NO, ConstantItem.EXCHANGE_BINANCE);
                    String largeKey = String.format(BinanceConstant.keyTemplate, val.getSymbol(), f, ConstantItem.ADJUST_NO, ConstantItem.EXCHANGE_BINANCE);
                    try {
                        FixedArrayList<Candlestick> smallAll = dbManager.getMdb().getBarMap().get(smallKey);
                        FixedArrayList<Candlestick> largeAll = dbManager.getMdb().getBarMap().get(largeKey);
                        if (null == smallAll) {
                            dbManager.getMdb().getBarMap().put(smallKey, new FixedArrayList<>(FixedArrayList.MAX_SIZE));
                            return;
                        }
                        if (null == largeAll) {
                            dbManager.getMdb().getBarMap().put(largeKey, new FixedArrayList<>(FixedArrayList.MAX_SIZE));
                            return;
                        }
                        List<Candlestick> diff = resample(val.getSymbol(), f);
                        if (largeAll.size() == 0) {
                            largeAll.addAll(diff);
                            single.put(largeKey, diff);
                            return;
                        }
                        Candlestick last = largeAll.get(largeAll.size() - 1);
                        // 无增量不更新
                        if (CollectionUtils.isEmpty(diff)) {
                            return;
                        }
                        if (last.getBarDate() < diff.get(0).getBarDate()) {
                            log.error("无法更新最后一根K线[{}]，数据可能存在断层", DateFormatUtils.format(new Date(last.getBarDate()), "YYYY-MM-dd HH:mm:ss"));
                            largeAll.addAll(diff);
                            single.put(largeKey, diff);
                        }
                        if (last.getBarDate().equals(diff.get(0).getBarDate())) {
                            largeAll.removeTail();
                            largeAll.addAll(diff);
                            single.put(largeKey, diff);
                        }
                        if (last.getBarDate() > diff.get(0).getBarDate()) {
                            largeAll.removeTail();
                            largeAll.addAll(diff.stream().filter(c -> c.getBarDate() >= last.getBarDate()).collect(Collectors.toList()));
                            single.put(largeKey, diff);
                        }
                    }catch (Exception e) {
                        log.error(String.format("【%s】重新采样失败", largeKey), e);
                    }
                });
                return single;
            }));
        });

        tasks.forEach(t -> {
            try {
                t.get();
            } catch (Exception e) {
                log.error("批量重新采样失败", e);
            }
        });

        log.error("batchResample，共{}条数据，耗时：{}ms", result.size(),System.currentTimeMillis() - start);
        return result;
    }

    public Long getCurrentBarTime() {
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.SECOND, 0);
        calendar.set(Calendar.MILLISECOND, 0);
        return calendar.getTimeInMillis();
    }

    private final static AtomicLong realLimit = new AtomicLong(0);

    /**
     * 持续发送回调
     */
    @Scheduled(cron = "${binance.barsCron}")
    public void scheduled() {
        log.info("K线增量更新开始....");
        // 重置频率限制
        realLimit.set(0);
        BinanceManager.totalLimit.set(0);

        if (dbManager.getState() != 2) {
            log.info("未初始化或者正在初始化....");
            return;
        }
        if (binanceProperties.getUseProxy() && client.getProxy() == null) {
            // 设置代理
            ProxyInfo proxySetting = binanceProperties.getProxies().get(0);
            Proxy proxyConn = new Proxy(Proxy.Type.HTTP, new InetSocketAddress(proxySetting.getHost(), proxySetting.getPort()));
            ProxyAuth proxy = new ProxyAuth(proxyConn, null);
            client.setProxy(proxy);
        }
        // 检测时间
        Long start = System.currentTimeMillis();
        Long time = JSON.parseObject(client.market().time()).getLong("serverTime");
        Long end = System.currentTimeMillis();
        log.error("获取时间耗时{}ms,本机时间{}, 交易所时间{}, 本机与交易所相差{}ms", end - start,
                DateFormatUtils.format(new Date(end), "YYYY-MM-dd HH:mm:ss,SSS"),
                DateFormatUtils.format(new Date(time), "YYYY-MM-dd HH:mm:ss,SSS"),
                end - time);


        // 1.检查1分钟的时间，最新时间如果相差超过一定的时间，则不进行爬取，否则会出现不连续的问题
        Long currentBarTime = getCurrentBarTime();
        String currentBarTimeStr = DateFormatUtils.format(new Date(currentBarTime), "yyyy-MM-dd HH:mm:ss");
        List<String> symbols = getNotLatestSymbols();
        Map<String, Future<List<Candlestick>>> tasks = new HashMap<>();
        Map<String, List<Candlestick>> result = new HashMap<>();

        // 创建任务
        String f = ConstantItem.FREQUENCY_1MIN;
//        RMap<String, Long> minTimeMap = redissonClient.getMap(String.format(minTime, f));
        Map<String,Long> minTimeMap = dbMapper.getTimeMap(f);
        long startTime = System.currentTimeMillis();
        int total = symbols.size();
//        AtomicInteger test = new AtomicInteger(0);
        while (symbols.size() > 0) {
            tasks.clear();
            for (String symbol:symbols) {
                // 先验证总频次
                Long val = BinanceManager.totalLimit.incrementAndGet();
                if (val > BinanceManager.MAX_LIMIT) {
                    log.error("本分钟采集超过总最大频率【{}】", BinanceManager.MAX_LIMIT);
                    break;
                }

                // 验证实时采集频次
                val = realLimit.incrementAndGet();
                if (val > BinanceManager.REAL_MAX_LIMIT) {
                    log.error("本分钟采集超过实时采集允许的最大频率【{}】", BinanceManager.REAL_MAX_LIMIT);
                    break;
                }

                // 在当前这一分钟内进行采集,过了50秒之后就没必要采集了，等待下一次任务
                if (new Date().getTime() - currentBarTime >= 50 * 1000) {
                    break;
                }

                Future<List<Candlestick>> future = pool.submit(() -> {
                    try {
//                        if (test.get() == 0 && symbol.equals("BTCUSDT")) {
//                            test.set(1);
//                            throw new RuntimeException("异常retry测试");
//                        }
                        LinkedHashMap<String, Object> parameters = new LinkedHashMap<>();
                        parameters.put("symbol", symbol);
                        parameters.put("interval", BinanceFrequency.getFrequency(f));
                        Long timestamp = minTimeMap.get(symbol);
                        parameters.put("startTime", timestamp);

                        String data = client.market().klines(parameters);
                        List<Candlestick> candlesticks = binanceCandlestickParser.parse(data);
                        return candlesticks;
                    } catch (BinanceConnectorException e) {
                        log.error(String.format("接口请求失败：【%s-%s】",symbol, f), e);
                        log.error("fullErrMessage: {}", e.getMessage(), e);
                    } catch (BinanceClientException e) {
                        log.error(String.format("接口请求失败：【%s-%s】", symbol, f), e);
                        log.error("fullErrMessage: {} \nerrMessage: {} \nerrCode: {} \nHTTPStatusCode: {}",
                                e.getMessage(), e.getErrMsg(), e.getErrorCode(), e.getHttpStatusCode(), e);
                    }catch (Exception e) {
                        log.error(String.format("采集异常：【%s-%s】", symbol, f), e);
                    }
                    return null;
                });
                tasks.put(symbol, future);
            }

//            long startTime = System.currentTimeMillis();
            List<String> unCompletedSymbols = new ArrayList<>();
            tasks.forEach((k, v) -> {
                try {
                    List<Candlestick> val = v.get();
                    // 用最新的值替换旧的数据
                    result.put(k, CollectionUtils.isNotEmpty(val) ? val : result.get(k));
                    // 未采集到当前的这一分钟的数据，需要重新采集
                    if (CollectionUtils.isEmpty(val) || !val.stream().filter(c -> c.getBarDate().equals(currentBarTime)).findFirst().isPresent()) {
                        unCompletedSymbols.add(k);
                    }
                } catch (Exception e) {
                    log.error(String.format("采集失败：【%s-%s】", k, f), e);
                }
            });

            symbols = unCompletedSymbols;
            if (symbols.size() > 0) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    log.error("线程休眠异常", e);
                }
            }
        }
        AtomicInteger completeCounter = new AtomicInteger(0);

        result.values().stream().forEach(r -> {
            if (r.stream().filter(c -> c.getBarDate().equals(currentBarTime)).findFirst().isPresent()) {
                completeCounter.incrementAndGet();
            }
        });
        log.error("采集【{}】数据采集完成，目标数：{}，完成数：{}，未完成数：{}, 耗时：{}ms",
                DateFormatUtils.format(new Date(currentBarTime), "yyyy-MM-dd HH:mm:ss,SSS"),
                total,
                completeCounter.get(),
                total - completeCounter.get(),
                System.currentTimeMillis() - startTime);

        startTime = System.currentTimeMillis();
        // 保存结果
        Map<String, List<Candlestick>> candleMap = new HashMap<>();
        result.forEach((k, v) -> {
            if (null == v) {
                return;
            }
            String key = String.format("binance:%s:%s:%s:%s", k, f, 3, 1);
            FixedArrayList<Candlestick> bars = dbManager.getMdb().getBarMap().get(key);
            Candlestick lastBar = bars.removeTail();
            if (null == lastBar) {
                return;
            }
            candleMap.put(key, v);
            List<Candlestick> latest = v.stream().filter(c -> c.getBarDate() >= lastBar.getBarDate()).collect(Collectors.toList());
            if (CollectionUtils.isNotEmpty(v)) {
                bars.addAll(latest);
            }
        });

        log.error("发送广播......");
        klineSubscribeMessageHandler.broadcast(candleMap);
        batchResample();
        log.error("在【{}】写入memory耗时：{}ms", DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss,SSS"), System.currentTimeMillis() - startTime);
    }



}
